package com.yungui.kafka.consumer.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yungui.kafka.consumer.constant.KafkaConstant;
import com.yungui.kafka.consumer.entity.FlatMessage;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 消息过滤器
 * author: hucheng
 */
@Component
public class FlatMessageFilter {

    /**
     * 过滤无效消息 有效消息: INSERT || UPDATE || DELETE
     *
     * @param messageList
     * @return
     */
    public List<FlatMessage> filterMessage(List<ConsumerRecord> messageList) {

        List<FlatMessage> effectMessageList = new ArrayList<>();
        for (ConsumerRecord record : messageList) {

            FlatMessage message = (FlatMessage) record.value();
            if (message == null || !KafkaConstant.EVENT_TYPES.contains(message.getType())) {
                // 无效消息，不处理。canal配置里了只处理DML binlog。
                continue;
            }
            effectMessageList.add(message);
        }
        return effectMessageList;
    }
}
